package cn.gwm.flink.streaming.dwd.check;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.dwd.function.ShuntStreamProcess;
import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

/**
 * @ClassName DwdHandleTemplate
 * @Description
 * @Author LiangGuang
 * @Date 2023/05/18 15:04
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public abstract class DwdHandleTemplate {
    public String uidSuffix = UUID.randomUUID().toString();

    public DataStream<JSONObject> handle(StreamExecutionEnvironment env) {
        OutputTag<JSONObject> errorStream = new OutputTag<JSONObject>("error") {
        };
        SingleOutputStreamOperator<JSONObject> dataStream = receiveAndHandle(env)
                .process(new ShuntStreamProcess(errorStream)).uid("process_" + uidSuffix)
                .map(item -> formatField(item)).uid("formatField_" + uidSuffix)
                .keyBy(item -> item.getStr(BaseFields.vin))
                .map(new LateDataFunction()).uid("delayHandle_" + uidSuffix);
        errorStreamHandle(dataStream.getSideOutput(errorStream));
        return dataStream;
    }

    /**
     * 错误数据的预留处理
     *
     * @param stream
     */
    public abstract void errorStreamHandle(DataStream<JSONObject> stream);

    /**
     * 接收数据并进行字段处置
     *
     * @return
     */
    public abstract DataStream<JSONObject> receiveAndHandle(StreamExecutionEnvironment env);

    /**
     * 格式化 字段名称 和补充
     *
     * @param source
     * @return
     */
    public abstract JSONObject formatField(JSONObject source);

}
